package com.shujia.spark.core

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
import scala.collection.mutable

object Demo26MapAndForeachPartitions {
  def main(args: Array[String]): Unit = {

    // mapPartitions、foreachPartition

    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName(this.getClass.getSimpleName.replace("$", ""))
    val sc = new SparkContext(conf)

    // 读取分数的文本数据 与 MySQL中的students数据进行关联，最后得到clazz总分top3

    val scoreRDD: RDD[String] = sc.textFile("spark/data/stu/score.txt")

    val sumScoreRDD: RDD[(String, Int)] = scoreRDD
      .map(line => (line.split(",")(0), line.split(",")(2).toInt))
      .reduceByKey(_ + _)

    // 第一种：将MySQL的数据读取过来再进行关联
    //创建数据连接对象//创建数据连接对象
    val conn: Connection = DriverManager.getConnection("jdbc:mysql://rm-bp1y7dm47j8h060vy4o.mysql.rds.aliyuncs.com:3307/bigdata32", "bigdata32", "123456")
    //获取hive操作对象//获取hive操作对象
    val pSt: PreparedStatement = conn.prepareStatement("select * from bigdata32.student")
    //执行sql语句//执行sql语句
    val resultSet: ResultSet = pSt.executeQuery()

    val stuMap: mutable.Map[String, (String, String)] = mutable.Map()

    while (resultSet.next()) {
      val id: String = resultSet.getString("id")
      val name: String = resultSet.getString("name")
      val clazz: String = resultSet.getString("clazz")
      stuMap.put(id, (name, clazz))
    }
    val stuMapBro: Broadcast[mutable.Map[String, (String, String)]] = sc.broadcast(stuMap)
    // 通过mapJoin进行关联
    sumScoreRDD
      .map(kv => {
        val nameAndClazz: (String, String) = stuMapBro.value.getOrElse(kv._1, ("", ""))
        (kv._1, nameAndClazz._1, nameAndClazz._2, kv._2)
      })
    //      .foreach(println)

    // 第二种方式：直接遍历总分数据，在遍历的同时像MySQL获取数据
    // 算子内部的代码必须全部实现序列化才能通过网络进行传输，MySQL的连接以及Statement都是不能被序列化的
    // 通过id直接去MySQL中查询该id的学生信息


    sumScoreRDD
      //      .map(kv=>{
      //        val id: String = kv._1
      //        // 创建数据连接对象
      //        // 在算子内部建立连接，相当于是在Task中直接建立连接，不需要再用到Driver建立好的连接，就没有序列化的问题了
      //        // 但由于map算子是每来一条数据就会执行一次，那就会导致每来一条数据就会建立一次连接，性能非常低
      //        val conn: Connection = DriverManager.getConnection("jdbc:mysql://rm-bp1y7dm47j8h060vy4o.mysql.rds.aliyuncs.com:3307/bigdata32", "bigdata32", "123456")
      //        val pSt02: PreparedStatement = conn.prepareStatement("select name,clazz from bigdata32.student where id = ?")
      //        pSt02.setString(1,id)
      //        val rs: ResultSet = pSt02.executeQuery()
      //        rs.next()
      //        val name: String = rs.getString("name")
      //        val clazz: String = rs.getString("clazz")
      //        (id,name,clazz,kv._2)
      //      })
      // mapPartitions适用于需要建立连接，然后从外部系统中获取数据，并进行后续的处理的场景
      // 连接建立的次数和RDD的分区数有关，让Task处理分区所有数据时可以公用一个连接，避免每一条数据建立一次连接
      .mapPartitions((it: Iterator[(String, Int)]) => {
        // 对RDD的每个分区进行依次的处理，该算子会将RDD的每个分区数据转换成Iterable迭代器
        // 如果需要对每条数据进行处理，则可以对迭代器进行遍历
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://rm-bp1y7dm47j8h060vy4o.mysql.rds.aliyuncs.com:3307/bigdata32", "bigdata32", "123456")
        val pSt02: PreparedStatement = conn.prepareStatement("select name,clazz from bigdata32.student where id = ?")

        // map方法并不是算子，它是迭代器的一个方法，所以没有算子内部外部之分，亦不会有序列化的问题
        it.map(kv => {
          val id: String = kv._1
          // 创建数据连接对象
          // 在算子内部建立连接，相当于是在Task中直接建立连接，不需要再用到Driver建立好的连接，就没有序列化的问题了
          // 但由于map算子是每来一条数据就会执行一次，那就会导致每来一条数据就会建立一次连接，性能非常低
          pSt02.setString(1, id)
          val rs: ResultSet = pSt02.executeQuery()
          rs.next()
          val name: String = rs.getString("name")
          val clazz: String = rs.getString("clazz")
          (id, name, clazz, kv._2)
        })
      })
      // 统计班级Top3
      .groupBy(t4 => t4._3)
      .flatMap(t2 => {
        t2._2.toList.sortBy(-_._4).take(3)
      })
      // ......统计步骤
      // foreachPartition：原理类似mapPartitions，区别在于是否需要做后续的处理
      // 适用于需要建立连接，将结果数据保存到外部系统中时
      .foreachPartition(i => {
        // 连接建立
        val conn: Connection = DriverManager.getConnection("jdbc:mysql://rm-bp1y7dm47j8h060vy4o.mysql.rds.aliyuncs.com:3307/bigdata32", "bigdata32", "123456")
        val pSt: PreparedStatement = conn.prepareStatement("insert into bigdata32.clazz_top3_zzk (id,name,clazz,sum_score) values (?,?,?,?)")

        var cnt: Int = 0
        i.foreach(t4 => {
          cnt += 1
          // 保存到MySQL中
          pSt.setString(1, t4._1)
          pSt.setString(2, t4._2)
          pSt.setString(3, t4._3)
          pSt.setInt(4, t4._4)

          // 将每个Statement加入到Batch批次中，等待一批一批执行
          pSt.addBatch()
          if (cnt == 1000) {
            println("执行了一次1000条数据的批量写入")
            pSt.executeBatch()
            cnt = 0 // 从零开始计数
          }
        })

        // 批量写入
        if (cnt != 0) {
          // 有可能第一次写入或者是最后一次写入数据量达不到一个批次的大小
          println("将最终未达到1000条数据的批次进行写入")
          pSt.executeBatch()
        }

      })


  }

}
